package com.px.framework.mqtt.handler;

import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Component
public class MqttHandlerSelector implements BeanPostProcessor {

    Map<String, List<IMqttHandler>> handlerMap = new HashMap<>();

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

        if (bean instanceof IMqttHandler) {
            IMqttHandler iMqttHandler = (IMqttHandler) bean;
            handlerMap.computeIfAbsent(iMqttHandler.bindTopicName().pattern(), v -> new ArrayList<>());
            handlerMap.computeIfPresent(iMqttHandler.bindTopicName().pattern(), (k, v) -> {
                v.add(iMqttHandler);
                return v;
            });
        }

        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }


    public void handler(String topicName, MqttMessage mqttMessage) {
        List<String> matchKeys = handlerMap.keySet().stream().filter(key -> Pattern.matches(key, topicName)).collect(Collectors.toList());
        matchKeys.parallelStream().forEach(key -> handlerMap.get(key).parallelStream().forEach(h -> h.handler(topicName, mqttMessage)));
    }

}
